package com.esunny.dns.ai.task;

import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class TaskQueueImpl implements TaskQueue, Runnable {

    private static final Log      log        = LogFactory.getLog(TaskQueueImpl.class);
    private Thread[]              threads    = null;
    private int                   queueSize  = 1000;
    private int                   workerSize = 1;
    private String                queneName  = "defaultQueue";
    private Queue<Task>           taskQueue  = new LinkedList<Task>();
    transient final ReentrantLock lock       = new ReentrantLock();
    transient final Condition     empty      = lock.newCondition();
    transient final Condition     full       = lock.newCondition();
    private volatile boolean      isExit     = false;

    public TaskQueueImpl(){
    }

    @Override
    public void enqueueTask(Task task) {
        final ReentrantLock locker = this.lock;
       
        locker.lock();
        try {
            while (taskQueue.size() > queueSize)
                try {
                    if (log.isInfoEnabled()) {
                        log.info("The queue is full,task size:" + taskQueue.size() + ",blocking enqueueTask");
                    }
                    empty.await();// P申请空资源
                } catch (InterruptedException e) {
                    log.error(e);
                }
            taskQueue.offer(task);
            full.signalAll();// V
        } finally {
            locker.unlock();
        }
    }

    @Override
    public void start() {
        threads = new Thread[workerSize];
        for (int i = 0; i < threads.length; i++) {
            (threads[i] = new Thread(this)).start();
            threads[i].setName(this.queneName + "-thread-" + String.valueOf(i));
        }
    }

    public void run() {
        while (!isExit) {
            Task task = null;
            final ReentrantLock locker = this.lock;
            locker.lock();
            try {
                while (taskQueue.size() == 0) {
                    try {
                        if (log.isInfoEnabled()) {
                            log.info("The queue is empty,task size:0 blocking work thread "
                                     + Thread.currentThread().getName());
                        }
                        full.await();// P 申请元素
                    } catch (InterruptedException e) {
                        log.error(e);
                    }
                }
                task = taskQueue.poll();
                empty.signalAll();// V
            } finally {
                locker.unlock();
            }
            if (task == null) return;
            try {
                long l = System.currentTimeMillis();
                task.execute();
                if (log.isInfoEnabled()) {
                    log.info("task " + task.toString() + " executed at:" + new Date() + ",consuming time:"
                             + (System.currentTimeMillis() - l) + " ms");
                }
            } catch (Exception ex) {
                log.error(ex);
            }
        }
    }

    /*
     * (non-Javadoc) 等待队列消耗完后正确退出，在调用stop后enqueue的任务将不再执行
     * @see com.esunny.dns.ai.task.TaskQueue#stop()
     */
    @Override
    public void stop() {
        if (log.isInfoEnabled()) {
            log.info("starting stop task queue,queue size:" + taskQueue.size());
        }
        for (@SuppressWarnings("unused")
        Thread worker : threads) {
            enqueueTask(null);
        }
        for (Thread worker : threads) {
            try {
                worker.join();
            } catch (InterruptedException e) {
                log.error(e);
            }
        }
        if (log.isInfoEnabled()) {
            log.info("successfully stop");
        }
    }

    /*
     * (non-Javadoc) 强制退出，工作线程将收到中断信号，除列中的任务将不再执行.
     * @see com.esunny.dns.ai.task.TaskQueue#abort()
     */
    @Override
    public void abort() {
        if (log.isInfoEnabled()) {
            log.info("starting abort task queue, queue size:" + taskQueue.size());
        }
        isExit = true;
        for (Thread t : threads) {
            try {
                t.interrupt();// 发出中断信号，退出组塞状态
            } catch (Exception ex) {
                log.error(ex);
            }
        }
        for (Thread t : threads) {
            try {
                t.join();
            } catch (Exception ex) {
                log.error(ex);
            }
        }
        final ReentrantLock locker = this.lock;
        locker.lock();
        try {
            empty.signalAll();
        } finally {
            locker.unlock();
        }

        if (log.isInfoEnabled()) {
            log.info("successfully abort");
        }
    }

    public int getQueueSize() {
        return queueSize;
    }

    public void setQueueSize(int queueSize) {
        if (queueSize <= 0) throw new IllegalArgumentException("queueSize must  greater then 0");
        this.queueSize = queueSize;
    }

    public int getWorkerSize() {
        return workerSize;
    }

    public void setWorkerSize(int workerSize) {
        if (workerSize <= 0) throw new IllegalArgumentException("workerSize must greater then 0");
        this.workerSize = workerSize;
    }

    public String getQueneName() {
        return queneName;
    }

    public void setQueneName(String queneName) {
        this.queneName = queneName;
    }

    public static void main(String[] args) throws InterruptedException {
        // final TaskQueueImpl task = new TaskQueueImpl();
        // task.setWorkerSize(10);
        // task.setQueueSize(10);
        // task.setQueneName("default");
        // Thread[] threads = new Thread[4];
        // task.start();
        // final AtomicInteger atomic = new AtomicInteger(1);
        // for (int i = 0; i < 4; i++) {
        // threads[i] = new Thread() {
        //
        // public void run() {
        // while (true) {
        // task.enqueueTask(new SleepTask("name-" + atomic.getAndIncrement()));
        // // try {
        // // Thread.sleep(1000);
        // // } catch (InterruptedException e) {
        // // // TODO Auto-generated catch block
        // // e.printStackTrace();
        // // }
        // }
        // }
        // };
        // threads[i].setDaemon(false);
        // threads[i].setName("p-thread" + String.valueOf(i));
        // }
        // for (Thread t : threads) {
        // t.start();
        // }
        //
        // Thread.sleep(1000 * 10);
        // task.abort();
        List<String> list=new LinkedList<String>();
        list.add("1");
        list.add("2");
        list.remove(0);
        System.out.println(list.get(0));
    }

    public static class SleepTask implements Task {

        private String name = null;

        public SleepTask(String name){
            this.name = name;
        }

        @Override
        public void execute() {
            System.out.println(name + "\n");
            // try {
            // Thread.sleep(1000);
            // } catch (InterruptedException e) {
            // // TODO Auto-generated catch block
            // e.printStackTrace();
            // }
        }
    }

}
